/*
 * 23/10/2011 - 15:07
 */
package org.nettyflux.mailbox;

import java.util.HashMap;
import java.util.HashSet;
import java.util.Map;
import java.util.Set;
import org.nettyflux.debug.GlobalDebugger;

/**
 * @author Victor Williams Stafusa da Silva
 */
public final class MailBox<T> {

    private final Set<ReceiveListener<T>> receiveListeners;
    private final Map<ReceiveListener<?>, ReceiveListener<T>> adaptedListeners;

    public MailBox() {
        this.receiveListeners = new HashSet<ReceiveListener<T>>(101);
        this.adaptedListeners = new HashMap<ReceiveListener<?>, ReceiveListener<T>>(101);
    }

    public synchronized void addReceiveListener(ReceiveListener<T> listener) {
        this.receiveListeners.add(listener);
    }

    @SuppressWarnings("element-type-mismatch")
    public synchronized void removeReceiveListener(ReceiveListener<? extends T> listener) {
        this.receiveListeners.remove(listener);
        ReceiveListener<T> casted = this.adaptedListeners.remove(listener);
        if (casted != null) this.receiveListeners.remove(casted);
    }

    public <A extends T, B extends A> void addReceiveListener(final Class<B> subclass, final ReceiveListener<A> listener) {
        ReceiveListener<T> castListener = new ReceiveListener<T>() {

            @Override
            public void receive(T message) {
                if (!subclass.isInstance(message)) return;
                listener.receive(subclass.cast(message));
            }
        };
        this.adaptedListeners.put(listener, castListener);
        this.receiveListeners.add(castListener);
    }

    public void postMessage(T message) {
        if (GlobalDebugger.isEnabled()) {
            GlobalDebugger.println("New message: " + message);
        }

        Set<ReceiveListener<T>> listeners;

        synchronized (this) {
            // Do a copy to avoid ConcurrentModificationException if some
            // listener registers or deregisters another listener (or itself)
            // during the iteration. Further, this allows the iteration
            // run outside the synchronized block.
            listeners = new HashSet<ReceiveListener<T>>(receiveListeners);
        }

        for (ReceiveListener<T> listener : listeners) {
            listener.receive(message);
        }
    }

    public T receiveBlocking(ReceiveAcceptor<T> acceptor) throws InterruptedException {
        NotifyingReceiveListener listener = new NotifyingReceiveListener(acceptor);
        addReceiveListener(listener);

        T accepted;
        synchronized (listener) {
            while ((accepted = listener.getAccepted()) == null) {
                listener.wait();
            }
        }

        return accepted;
    }

    private class NotifyingReceiveListener implements ReceiveListener<T> {

        private final ReceiveAcceptor<T> acceptor;
        private volatile T accepted;

        public NotifyingReceiveListener(ReceiveAcceptor<T> acceptor) {
            this.acceptor = acceptor;
        }

        public T getAccepted() {
            return accepted;
        }

        @Override
        public synchronized void receive(T message) {
            if (accepted != null) return;
            if (!acceptor.accept(message)) return;

            accepted = message;
            removeReceiveListener(this);
            notify();
        }
    }
}
